package com.xx.hdfs;

import org.apache.hadoop.fs.*;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class HdfsDemo {
    FileSystem fs;

    public HdfsDemo(FileSystem fs) {
        this.fs = fs;
    }

    public boolean uploadFile(Path src, Path dfs) {
        try {
            if (exists(new Path(dfs.toString() + "/" + src.getName()))) {
                System.out.println("dfs is already exists");
                return false;
            }
            File local = new File(src.toString());
            if (local.isFile()) fs.copyFromLocalFile(src, dfs);
            else {
                File[] files = local.listFiles();
                Set<File> set = new HashSet<>();
                assert files != null;
                for (File file : files) uploadAdd(set, file);
                File srcFile = new File(src.toString());
                set.add(srcFile);
                int index = src.toString().indexOf(src.getName());
                ArrayList<File> collect = set.stream().sorted(Comparator.comparingInt(e -> e.getPath().length())).collect(Collectors.toCollection(ArrayList::new));
                collect.forEach(e -> {
                    Path path = new Path(dfs + "/" + e.getPath().substring(index).replace(File.separator, "/"));
                    if (e.isDirectory()) mkdirs(path);
                    else {
                        try {
                            fs.copyFromLocalFile(new Path(e.getPath()), path);
                        } catch (IOException ex) {
                            throw new RuntimeException(ex);
                        }
                    }
                });
            }
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    private void uploadAdd(Set<File> set, File file) {
        set.add(file);
        if (file.isDirectory())
            for (File listFile : Objects.requireNonNull(file.listFiles())) uploadAdd(set, listFile);
    }

    public boolean download(Path src, Path dfs) {
        try {
            if (new File(src + File.separator + dfs.getName()).exists()) {
                System.out.println("this file is exists");
                return false;
            }
            if (fs.isFile(dfs)) fs.copyToLocalFile(false, dfs, new Path(src + File.separator + dfs.getName()), true);
            else {
                RemoteIterator<LocatedFileStatus> files = fs.listLocatedStatus(dfs);
                Set<FileStatus> fileSet = new HashSet<>();
                fileSet.add(getFileStatus(dfs));
                while (files.hasNext()) downloadAdd(files.next(), fileSet);
                LinkedHashSet<FileStatus> collect =
                        fileSet.stream().sorted(Comparator.comparingInt(e -> e.getPath().toString().length())).collect(Collectors.toCollection(LinkedHashSet::new));
                int index = dfs.toString().indexOf(dfs.getName());

                collect.forEach(e -> {
                    File localFile = new File(src + File.separator + e.getPath().toString().substring(index));
                    if (e.isFile()) {
                        try {
                            fs.copyToLocalFile(false, e.getPath(),
                                    new Path(localFile.getPath()),
                                    true);
                        } catch (IOException ex) {
                            throw new RuntimeException(ex);
                        }
                    } else localFile.mkdirs();
                });
            }
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    public void downloadAdd(FileStatus status, Set<FileStatus> set) throws IOException {
        set.add(status);
        if (status.isDirectory()) {
            RemoteIterator<LocatedFileStatus> fileStatuses = fs.listLocatedStatus(status.getPath());
            while (fileStatuses.hasNext()) downloadAdd(fileStatuses.next(), set);
        }
    }

    public FSDataOutputStream create(Path path, boolean overwrite) {
        try {
            return fs.create(path, overwrite);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean rename(Path old, Path newPath) {
        try {
            for (FileStatus fileStatus : fs.listStatus(newPath.getParent())) {
                if (fileStatus.getPath().getName().equals(newPath.getName()))
                    return false;
            }
            return fs.rename(old, newPath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean exists(Path p) {
        try {
            return fs.exists(p);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean mkdirs(Path p) {
        try {
            return fs.mkdirs(p);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public FileStatus getFileStatus(Path p) {
        try {
            return fs.getFileStatus(p);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean createNewFile(Path p) {
        try {
            return fs.createNewFile(p);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean delete(Path p, boolean b) {
        try {
            return fs.delete(p, b);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public FSDataInputStream open(Path path) {
        try {
            return fs.open(path);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void close() {
        try {
            fs.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public boolean test() {
        boolean exists = exists(new Path("/"));
        close();
        return exists;
    }
}
